package org.hzandyy.zookeeper.demo;

import lombok.extern.slf4j.Slf4j;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.ZooDefs;
import org.hzandyy.zookeeper.util.KeeperAsyncUtils;

import java.util.stream.Stream;

@Slf4j
public class Master extends BaseConnection {

    private volatile MasterStates state;

    public Master() {
        watcher = event -> {
            // Watcher.Event.EventType.None 表明为会话状态
            if (event.getType() == Watcher.Event.EventType.None) {
                switch (event.getState()) {
                    case SyncConnected:
                        connected = true;
                        break;
                    case Disconnected:
                        connected = false;
                        break;
                    case Expired:
                        expired = true;
                        connected = false;
                        log.error("session expiration");
                    default:
                        break;
                }
            }
        };
    }

    // ************** 创建基础功能节点 ***************

    public void bootstrap() {
        String[] nodes = new String[]{
            "/workers", // 用于存放从节点
            "/assign", // 用于分配任务
            "/tasks", // 用于客户端提交任务
            "/status" // 记录任务状态
        };
        Stream.of(nodes)
            .forEach(znode -> createParent(znode, emptyContent));
    }

    private void createParent(String znode, byte[] data) {
        zooKeeper.create(znode, data, ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT,
            (rc, path, ctx, name) -> {
                switch (KeeperAsyncUtils.code(rc)) {
                    case CONNECTIONLOSS:
                        createParent(znode, (byte[]) ctx);
                        break;
                    case OK:
                        log.info("基础功能节点创建成功 === [{}]", path);
                        break;
                    case NODEEXISTS:
                        log.warn("基础功能节点已存在 === [{}]", path);
                        break;
                    default:
                        log.error("创建基础功能节点失败 === {}", KeeperAsyncUtils.exception(rc, path));
                }
            }
            , data);
    }

    // ******************* 选举群首 ****************

    enum MasterStates {
        RUNNING,
        ELECTED,
        NOTELECTED
    }
}
